package cn.finder.producer;

import cn.finder.Config;
import org.apache.kafka.clients.producer.*;

import java.util.Date;
import java.util.Properties;
import java.util.concurrent.Future;

// 生产者Demo
// 用在 下单服务中
public class ProducerDemo
{
    public static void main( String[] args ){


        KafkaProducer producer = new KafkaProducer<Integer, String>(buildProperties());
        EntrustOrder order=buildOrder();

    //异步回调方式
        /*

        producer.send(new ProducerRecord<Long, String>("entrust_" + order.getSymbol(), order.getEntrust_id(), order.toJson()), new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {

            }
        });*/


        // 同步方式: 推荐使用   因为下单需要当时告知操作状态
       Future<RecordMetadata> future= producer.send(new ProducerRecord<Long, String>("entrust_"+order.getSymbol(), order.getEntrust_id(),order.toJson()));
       try {
           RecordMetadata metadata = future.get(); //阻塞; 如果不执行 get  则异步，通过callback回调
            System.out.println("下单成功...");
       }
       catch (Exception e){
           System.out.println("下单失败处理...");
           e.printStackTrace();

       }



        // App exit
       producer.close();

    }

    private static Properties buildProperties(){
        Properties props = new Properties();
        props.put("bootstrap.servers", Config.Kafka_host);

        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);

        props.put("key.serializer", "org.apache.kafka.common.serialization.LongSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        return props;
    }

    private static EntrustOrder buildOrder(){
        EntrustOrder order=new EntrustOrder();
        order.setSymbol("aaa_usdt");
        order.setBuy_sell("S");
        order.setEntrust_type("L");
        order.setAmt(3.0);
        order.setPrice(1.0);
        order.setMaker_charge(0.00018);
        order.setTaker_charge(0.0002);
        order.setSource("web");
        order.setMarket_type("S"); //现货

        order.setEntrust_id(new Date().getTime());// 根据业务生成唯一id
        return order;
    }
}
